package com.xxl.job.executor.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xxl.job.core.log.XxlJobLogger;
import com.xxl.job.executor.common.util.SpringContextHolder;
import com.xxl.job.executor.constant.Constants;
import com.xxl.job.executor.enums.StatusCodeEnum;
import com.xxl.job.executor.exception.BusinessException;
import com.xxl.job.executor.httpClient.core.HttpClientUtils;
import com.xxl.job.executor.persistence.beans.*;
import com.xxl.job.executor.pojo.*;
import com.xxl.job.executor.service.IExceptionDistributeService;
import com.xxl.job.executor.util.KafkaUtil;
import lombok.Data;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.http.HttpResponse;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.sql.DataSource;
import java.io.*;
import java.sql.Connection;
import java.util.*;
import java.util.concurrent.*;

import static com.xxl.job.executor.util.GZipUtils.compress;

/**
 * 异常流水回调服务
 */
@Service("exceptionDistributeService")
public class ExceptionDistributeService implements IExceptionDistributeService {
    @Value("${kafka.bootstrap.servers}")
    private String kafkaBS;

    @Value("${kafka.key.serializer}")
    private String kafkaKS;

    @Value("${kafka.value.serializer}")
    private String kafkaVS;

    @Value("${httpclient.url}")
    private String middlegroundUrl;

    private final Integer corePoolSize = 5;
    private final Integer maxPoolSize = 20;


    /**
     * MTC入口流水回调
     * @throws Exception
     */
    public void callback_binList() throws Exception{
        XxlJobLogger.log("=====开始执行业务操作===");
        DataSource dataSourcePre = SpringContextHolder.getBean("dataSourcePre");
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSourcePre);
        Connection conn = null;
        try {
            conn = jdbcTemplate.getDataSource().getConnection();
            conn.setAutoCommit(false);

            String querySql = "select * from Binlist_err where haddeal in(1,3,4) and rownum <= 2000";
            List<BInListErrEntity> list = jdbcTemplate.query(querySql.toString(), new BeanPropertyRowMapper(BInListErrEntity.class));
            XxlJobLogger.log("[普通]回调异常数据共(条):" + list.size());

            //批量更新发送状态sql
            String batchDelSql = "delete from Binlist_err where haddeal in(1,3,4) and listNo = ?";

            XxlJobLogger.log("[业务]待发送业务系统流水数:" + list.size());
            List<Object[]> listNos = new ArrayList<Object[]>();
            BInListErrPojo pojo = null;
            for(int i = 0; i < list.size(); i++){
                BInListErrEntity entity = list.get(i);
                String listNo = entity.getListno();//流水号
                listNos.add(new String[]{listNo});
            }

            if (listNos.size() > 0){
                //将异常流水表中的记录插入到相应的已修正表
                try{
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_DEL_CHARGE_BINLIST), listNos);
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_INST_CHARGE_BINLIST), listNos);
                }catch (Exception e){
                    e.printStackTrace();
                    XxlJobLogger.log("插入已发送异常流水表出错:"+listNos.toString());
                    throw e;
                }
                jdbcTemplate.batchUpdate(batchDelSql, listNos);//删除异常流水表中的记录
                conn.commit();//手动提交
            }

            XxlJobLogger.log("[发送]开始发送异常流水:" + list.size());

            KafkaProducer producer = KafkaUtil.getKafkaProducer(kafkaBS,kafkaKS,kafkaVS);
            ThreadPoolExecutor tpe = new ThreadPoolExecutor(corePoolSize, maxPoolSize, 0,
                    TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
                    //,new ThreadPoolExecutor.CallerRunsPolicy());
            List<Future<String>> futureList = new ArrayList<>();
            //存储线程的返回值
            List<Object[]> failListNos = new ArrayList<>();
            for (int i = 0; i < list.size(); i++) {
                BInListErrEntity entity = list.get(i);
                if(1 == entity.getHaddeal() || 3 == entity.getHaddeal()) {//只有为 1-已治理 状态时才进行发送
                    String listNo = entity.getListno();
                    String jsonName = entity.getJsonname();//包名
                    short seq = entity.getSeq();

                    //复制对象到用于发送kafka的对象中
                    pojo = new BInListErrPojo();
                    BeanUtils.copyProperties(entity, pojo);

                    //将bean对象转换成json对象用于发送kafka
                    String jsonStr = JSON.toJSONString(pojo);
                    jsonStr = JSON.toJSONString(JSONObject.parseObject(jsonStr, BInList.class));
                    JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                    JSONArray jsonArray = new JSONArray();
                    jsonArray.add(jsonObject);


                    Task task = new Task(producer,listNo,jsonArray,getJsonname(jsonName, seq));

                    //调用submit可以获得线程的返回值
                    Future<String> result = tpe.submit(task);
                    futureList.add(result);
                }
            }
            //closeKafkaProducer(producer);

            for(Future<String> result : futureList){
                if(!"success".equals(result.get())) {
                    failListNos.add(new Object[]{result.get()});
                }
            }
            tpe.shutdown();

            //将发送失败记录的haddeal状态更新为 1-已治理
            jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_UPD_CHARGE_BINLIST), failListNos);

            XxlJobLogger.log("[普通]已治理状态记录发送失败(条):" + failListNos.size());
            XxlJobLogger.log("=====完成执行业务操作===");
        }catch(Exception e){
            if (conn != null) {
                try {
                    conn.rollback();
                    XxlJobLogger.log("回滚成功.");
                }catch (Exception e1){
                    e1.printStackTrace();
                }
            }
            XxlJobLogger.log("抛出异常："+e.getMessage());
            throw e;
        }finally{
            if (conn != null) {
                try {
                    conn.setAutoCommit(true);
                    conn.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * ETC入口流水回调
     * @throws Exception
     */
    public void callback_etcbinList() throws Exception{
        XxlJobLogger.log("=====开始执行业务操作===");
        DataSource dataSourcePre = SpringContextHolder.getBean("dataSourcePre");
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSourcePre);
        Connection conn = null;
        try {
            conn = jdbcTemplate.getDataSource().getConnection();
            conn.setAutoCommit(false);

            String querySql = "select * from EtcBinlist_err where haddeal in(1,3,4) and rownum <= 2000";
            List<ETCBInListErrEntity> list = jdbcTemplate.query(querySql.toString(), new BeanPropertyRowMapper(ETCBInListErrEntity.class));
            XxlJobLogger.log("[普通]回调异常数据共(条):" + list.size());

            //批量更新发送状态sql
            String batchDelSql = "delete from EtcBinlist_err where haddeal in(1,3,4) and listNo = ?";

            XxlJobLogger.log("[业务]待发送业务系统流水数:"+list.size());
            List<Object[]> listNos = new ArrayList<Object[]>();
            ETCBInListErrPojo pojo = null;
            for(int i = 0; i < list.size(); i++){
                ETCBInListErrEntity entity = list.get(i);
                String listNo = entity.getListno();//流水号
                listNos.add(new String[]{listNo});
            }
            //针对少于 500条时的处理
            if(listNos.size() > 0){
                //将异常流水表中的记录插入到相应的已修正表
                try{
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_DEL_CHARGE_ETCBINLIST), listNos);
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_INST_CHARGE_ETCBINLIST), listNos);
                }catch (Exception e){
                    e.printStackTrace();
                    XxlJobLogger.log("插入已发送异常流水表出错:"+listNos.toString());
                    throw e;
                }
                jdbcTemplate.batchUpdate(batchDelSql,listNos);
            }
            conn.commit();//手动提交

            XxlJobLogger.log("[发送]开始发送异常流水:" + list.size());

            KafkaProducer producer = KafkaUtil.getKafkaProducer(kafkaBS,kafkaKS,kafkaVS);
            ThreadPoolExecutor tpe = new ThreadPoolExecutor(corePoolSize, maxPoolSize, 0,
                    TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());//,new ThreadPoolExecutor.CallerRunsPolicy());
            //存储线程的返回值
            List<Future<String>> futureList = new ArrayList<>();
            List<Object[]> failListNos = new ArrayList<Object[]>();
            for(int i = 0; i < list.size(); i++){
                ETCBInListErrEntity entity = list.get(i);
                String listNo = entity.getListno();//流水号
                String jsonName = entity.getJsonname();//包名
                short seq = entity.getSeq();

                //复制对象到用于发送kafka的对象中
                pojo = new ETCBInListErrPojo();
                BeanUtils.copyProperties(entity,pojo);

                if(1 == entity.getHaddeal() || 3 == entity.getHaddeal()) {//只有为 1-已治理 状态时才进行发送
                    //将bean对象转换成json对象用于发送kafka
                    String jsonStr = JSON.toJSONString(pojo);
                    jsonStr = JSON.toJSONString(JSONObject.parseObject(jsonStr, ETCBInList.class));
                    JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                    JSONArray jsonArray = new JSONArray();
                    jsonArray.add(jsonObject);

                    Task task = new Task(producer,listNo,jsonArray,getJsonname(jsonName, seq));
                    //调用submit可以获得线程的返回值
                    Future<String> result = tpe.submit(task);
                    futureList.add(result);
                }
            }
            //closeKafkaProducer(producer);

            for(Future<String> result : futureList){
                if(!"success".equals(result.get())) {
                    failListNos.add(new Object[]{result.get()});
                }
            }
            tpe.shutdown();

            //将发送失败记录的haddeal状态更新为 1-已治理
            jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_UPD_CHARGE_ETCBINLIST), failListNos);
            XxlJobLogger.log("[普通]已治理状态记录发送失败(条):" + failListNos.size());
            XxlJobLogger.log("=====完成执行业务操作===");
        }catch(Exception e){
            if (conn != null) {
                try {
                    conn.rollback();
                    XxlJobLogger.log("回滚成功.");
                }catch (Exception e1){
                    e1.printStackTrace();
                }
            }
            XxlJobLogger.log("抛出异常："+e.getMessage());
            throw e;
        }finally{
            if (conn != null) {
                try {
                    conn.setAutoCommit(true);
                    conn.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * MTC出口流水回调
     * @throws Exception
     */
    public void callback_boutList() throws Exception{
        XxlJobLogger.log("=====开始执行业务操作===");
        DataSource dataSourcePre = SpringContextHolder.getBean("dataSourcePre");
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSourcePre);
        Connection conn = null;
        try {
            conn = jdbcTemplate.getDataSource().getConnection();
            conn.setAutoCommit(false);

            String querySql = "select * from Boutlist_err where haddeal in(1,3,4) and rownum <= 2000";
            List<BoutListErrEntity> list = jdbcTemplate.query(querySql.toString(), new BeanPropertyRowMapper(BoutListErrEntity.class));
            XxlJobLogger.log("[普通]回调异常数据共(条):" + list.size());

            //批量更新发送状态sql
            NamedParameterJdbcTemplate nameJdbc = new NamedParameterJdbcTemplate(jdbcTemplate);
            String batchDelDetailSql = "delete from Boutlist_detail_err where haddeal in(1,3,4) and listNo like ?";
            String batchDelSql = "delete from Boutlist_err where haddeal in(1,3,4) and listNo = ?";

            List<Object[]> listNos = new ArrayList<Object[]>();
            Map<String,BoutListErrPojo> sendMap = new HashMap<String,BoutListErrPojo>();
            XxlJobLogger.log("[业务]待发送业务系统流水数:"+list.size());
            for(int i = 0; i < list.size(); i++){
                BoutListErrEntity entity = list.get(i);
                String jsonName = entity.getJsonname();//包名
                short seq = (short)entity.getSeq();
                String listNo = entity.getListno();//流水号

                //复制对象到用于发送kafka的对象中
                List<BoutListDetailErrPojo> pojoList = null;
                BoutListDetailErrPojo detailPojo = null;

                String detailSql = "select * from BOUTLIST_DETAIL_ERR where haddeal in(1,3,4) and listNo like :listno order by listno asc";
                Map<String,Object> params = new HashMap<String,Object>();
                params.put("listno",listNo+"%");
                List<BoutListDetailErrEntity> detailList = nameJdbc.query(detailSql,params,new BeanPropertyRowMapper(BoutListDetailErrEntity.class));
                if(detailList.size() > 0) {
                    pojoList = new ArrayList<BoutListDetailErrPojo>();
                    for (int j = 0; j < detailList.size(); j++) {
                        BoutListDetailErrEntity detailErrEntity = detailList.get(j);

                        detailPojo = new BoutListDetailErrPojo();
                        BeanUtils.copyProperties(detailErrEntity, detailPojo);
                        pojoList.add(detailPojo);
                    }
                }

                //组装最终对象
                BoutListErrPojo pojo = new BoutListErrPojo();
                BeanUtils.copyProperties(entity,pojo);
                pojo.setDataDtail(pojoList);//将子明细放进对象中

                //将待发送对象加入到map中
                if(1 == entity.getHaddeal() || 3 == entity.getHaddeal()) {//只有为 1-已治理 状态时才进行发送
                    sendMap.put(getJsonname(jsonName, seq), pojo);
                }

                listNos.add(new String[]{listNo});
            }

            if(listNos.size() > 0){
                //批量插入
                try {
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_DEL_CHARGE_BOUTLIST), listNos);
                    int[] res = jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_INST_CHARGE_BOUTLIST), listNos);
                }catch (Exception e){
                    e.printStackTrace();
                    XxlJobLogger.log("插入已发送异常流水表出错:"+listNos.toString());
                    throw e;
                }
                jdbcTemplate.batchUpdate(batchDelSql,listNos);//批量删除

                //根据流水号模糊匹配删除明细记录
                List<Object[]> detailListNos = new ArrayList<Object[]>();
                for(Object[] obj : listNos){
                    String _ListNo = String.valueOf(obj[0]);
                    detailListNos.add(new Object[]{_ListNo.concat("%")});
                }
                //根据流水号模糊匹配插入明细记录
                try{
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_DEL_CHARGE_BOUTLIST_DETAIL),detailListNos);
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_INST_CHARGE_BOUTLIST_DETAIL),detailListNos);
                }catch (Exception e){
                    e.printStackTrace();
                    XxlJobLogger.log("插入已发送异常流水明细表出错:"+detailListNos.toString());
                    throw e;
                }
                jdbcTemplate.batchUpdate(batchDelDetailSql,detailListNos);//批量删除异常明细记录
            }
            conn.commit();//手动提交

            XxlJobLogger.log("[发送]开始发送异常流水:" + list.size());
            KafkaProducer producer = KafkaUtil.getKafkaProducer(kafkaBS,kafkaKS,kafkaVS);
            ThreadPoolExecutor tpe = new ThreadPoolExecutor(corePoolSize, maxPoolSize, 0,
                    TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());//,new ThreadPoolExecutor.CallerRunsPolicy());
            //存储线程的返回值
            List<Future<String>> futureList = new ArrayList<>();
            List<Object[]> failListNos = new ArrayList<Object[]>();
            for(String jsonName : sendMap.keySet()){
                //将bean对象转换成json对象用于发送kafka
                BoutListErrPojo pojo = sendMap.get(jsonName);
                String listNo = pojo.getListno();
                String jsonStr = JSON.toJSONString(pojo);
                jsonStr = JSON.toJSONString(JSONObject.parseObject(jsonStr, BOutList.class));
                JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                JSONArray jsonArray = new JSONArray();
                jsonArray.add(jsonObject);

                Task task = new Task(producer,listNo,jsonArray,jsonName);
                //调用submit可以获得线程的返回值
                Future<String> result = tpe.submit(task);
                futureList.add(result);
            }
           // closeKafkaProducer(producer);

            for(Future<String> result : futureList){
                if(!"success".equals(result.get())) {
                    failListNos.add(new Object[]{result.get()});
                }
            }
            tpe.shutdown();

            //将发送失败记录的haddeal状态更新为 1-已治理
            jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_UPD_CHARGE_BOUTLIST), failListNos);
            conn.commit();//手动提交

            XxlJobLogger.log("[普通]已治理状态记录发送失败(条):" + failListNos.size());
            XxlJobLogger.log("=====完成执行业务操作===");
        }catch(Exception e){
            if (conn != null) {
                try {
                    conn.rollback();
                    XxlJobLogger.log("回滚成功.");
                }catch (Exception e1){
                    e1.printStackTrace();
                }
            }
            XxlJobLogger.log("抛出异常："+e.getMessage());
            throw e;
        }finally{
            if (conn != null) {
                try {
                    conn.setAutoCommit(true);
                    conn.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * ETC出口流水回调
     * @throws Exception
     */
    public void callback_etcboutList() throws Exception{
        XxlJobLogger.log("=====开始执行业务操作===");
        DataSource dataSourcePre = SpringContextHolder.getBean("dataSourcePre");
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSourcePre);
        Connection conn = null;
        try {
            conn = jdbcTemplate.getDataSource().getConnection();
            conn.setAutoCommit(false);

            String querySql = "select * from EtcBoutlist_err where haddeal in(1,3,4) and rownum <= 2000";
            List<ETCBOutListErrEntity> list = jdbcTemplate.query(querySql.toString(), new BeanPropertyRowMapper(ETCBOutListErrEntity.class));
            XxlJobLogger.log("[普通]回调异常数据共(条):" + list.size());

            //批量更新发送状态sql
            NamedParameterJdbcTemplate nameJdbc = new NamedParameterJdbcTemplate(jdbcTemplate);
            String batchDelDetailSql = "delete from EtcBoutlist_detail_err where haddeal in(1,3,4) and listNo like ?";
            String batchDelSql = "delete from EtcBoutlist_err where haddeal in(1,3,4) and listNo = ?";

            XxlJobLogger.log("[业务]待发送业务系统流水数:"+list.size());
            List<Object[]> listNos = new ArrayList<Object[]>();
            Map<String,ETCBOutListErrPojo> sendMap = new HashMap<String,ETCBOutListErrPojo>();
            for(int i = 0; i < list.size(); i++){
                ETCBOutListErrEntity entity = list.get(i);
                String jsonName = entity.getJsonname();//包名
                String listNo = entity.getListno();//流水号
                short seq = (short)entity.getSeq();

                //复制对象到用于发送kafka的对象中
                List<ETCBoutListDetailErrPojo> pojoList = null;
                ETCBoutListDetailErrPojo detailPojo = null;

                String detailSql = "select * from ETCBOUTLIST_DETAIL_ERR where haddeal in(1,3,4) and listNo like :listno order by listno asc";
                Map<String,Object> params = new HashMap<String,Object>();
                params.put("listno",listNo+"%");
                List<ETCBoutListDetailErrEntity> detailList = nameJdbc.query(detailSql,params,new BeanPropertyRowMapper(ETCBoutListDetailErrEntity.class));
                if(detailList.size() > 0) {
                    pojoList = new ArrayList<ETCBoutListDetailErrPojo>();
                    for (int j = 0; j < detailList.size(); j++) {
                        ETCBoutListDetailErrEntity detailErrEntity = detailList.get(j);

                        detailPojo = new ETCBoutListDetailErrPojo();
                        BeanUtils.copyProperties(detailErrEntity, detailPojo);
                        pojoList.add(detailPojo);
                    }
                }

                //组装最终对象
                ETCBOutListErrPojo pojo = new ETCBOutListErrPojo();
                BeanUtils.copyProperties(entity,pojo);
                pojo.setDataDtail(pojoList);//将子明细放进对象中
                if(1 == entity.getHaddeal() || 3 == entity.getHaddeal()) {//只有为 1-已治理 状态时才进行发送
                    sendMap.put(jsonName,pojo);
                }

                listNos.add(new String[]{listNo});
            }


            if(listNos.size() > 0){
                try{
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_DEL_CHARGE_ETCBOUTLIST),listNos);
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_INST_CHARGE_ETCBOUTLIST),listNos);
                }catch (Exception e){
                    e.printStackTrace();
                    XxlJobLogger.log("插入已发送异常流水表出错:"+listNos.toString());
                    throw e;
                }
                jdbcTemplate.batchUpdate(batchDelSql,listNos);//删除主流水表记录

                //根据流水号模糊匹配删除明细记录
                List<Object[]> detailListNos = new ArrayList<Object[]>();
                for(Object[] obj : listNos){
                    String _ListNo = String.valueOf(obj[0]);
                    detailListNos.add(new Object[]{_ListNo.concat("%")});
                }

                try{
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_DEL_CHARGE_ETCBOUTLIST_DETAIL),detailListNos);//插入明细表记录
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_INST_CHARGE_ETCBOUTLIST_DETAIL),detailListNos);//插入明细表记录
                }catch (Exception e){
                    e.printStackTrace();
                    XxlJobLogger.log("插入已发送异常流水明细表出错:"+detailListNos.toString());
                    throw e;
                }
                jdbcTemplate.batchUpdate(batchDelDetailSql,detailListNos);//删除明细表记录
            }
            conn.commit();//手动提交

            XxlJobLogger.log("[发送]开始发送异常流水:" + list.size());
            KafkaProducer producer = KafkaUtil.getKafkaProducer(kafkaBS,kafkaKS,kafkaVS);
            ThreadPoolExecutor tpe = new ThreadPoolExecutor(corePoolSize, maxPoolSize, 0,
                    TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());//,new ThreadPoolExecutor.CallerRunsPolicy());
            List<Future<String>> futureList = new ArrayList<>();
            List<Object[]>  failListNos = new ArrayList<Object[]>();
            for(String jsonName : sendMap.keySet()){
                //将bean对象转换成json对象用于发送kafka
                ETCBOutListErrPojo pojo = sendMap.get(jsonName);
                String listNo = pojo.getListno();
                String jsonStr = JSON.toJSONString(pojo);
                jsonStr = JSON.toJSONString(JSONObject.parseObject(jsonStr, ETCBOutList.class));
                JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                JSONArray jsonArray = new JSONArray();
                jsonArray.add(jsonObject);

                Task task = new Task(producer,listNo,jsonArray,jsonName);
                //调用submit可以获得线程的返回值
                Future<String> result = tpe.submit(task);
                futureList.add(result);
            }
            //closeKafkaProducer(producer);

            for(Future<String> result : futureList){
                if(!"success".equals(result.get())) {
                    failListNos.add(new Object[]{result.get()});
                }
            }
            tpe.shutdown();

            //将发送失败记录的haddeal状态更新为 1-已治理
            jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_UPD_CHARGE_ETCBOUTLIST),failListNos);
            conn.commit();//手动提交

            XxlJobLogger.log("已治理状态记录发送失败(条):" + failListNos.size());
            XxlJobLogger.log("=====完成执行业务操作===");
        }catch(Exception e){
            if (conn != null) {
                try {
                    conn.rollback();
                    XxlJobLogger.log("回滚成功.");
                }catch (Exception e1){
                    e1.printStackTrace();
                }
            }
            XxlJobLogger.log("抛出异常："+e.getMessage());
            throw e;
        }finally{
            if (conn != null) {
                try {
                    conn.setAutoCommit(true);
                    conn.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 门架流水回调
     * @throws Exception
     */
    public void callback_etctuList() throws Exception{
        XxlJobLogger.log("=====开始执行业务操作===");
        DataSource dataSourcePre = SpringContextHolder.getBean("dataSourcePre");
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSourcePre);
        Connection conn = null;
        try {
            conn = jdbcTemplate.getDataSource().getConnection();
            conn.setAutoCommit(false);

            String querySql = "select * from grantry_trans_main_ERR where haddeal in(1,3,4) and rownum < 2000";
            List<GrantryTransMainErrEntity> list = jdbcTemplate.query(querySql.toString(), new BeanPropertyRowMapper(GrantryTransMainErrEntity.class));
            XxlJobLogger.log("[普通]回调异常数据共(条):" + list.size());

            //批量更新发送状态sql
            String batchUpdSql = "delete grantry_trans_main_ERR where haddeal in(1,3,4) and tradeId = ?";

            List<Object[]> tradeIds = new ArrayList<Object[]>();
            GrantryTransMainErrPojo pojo = null;
            XxlJobLogger.log("[业务]待发送业务系统流水数:"+list.size());
            for(int i = 0; i < list.size(); i++){
                GrantryTransMainErrEntity entity = list.get(i);
                String tradeid = entity.getTradeid();//交易号

                tradeIds.add(new String[]{tradeid});
            }

            if(tradeIds.size() > 0){
                try{
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_DEL_CHARGE_GRANTRY_MAIN),tradeIds);
                    jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_INST_CHARGE_GRANTRY_MAIN),tradeIds);
                }catch (Exception e){
                    e.printStackTrace();
                    XxlJobLogger.log("插入已发送异常流水表出错:"+tradeIds.toString());
                    throw e;
                }
                jdbcTemplate.batchUpdate(batchUpdSql,tradeIds);
            }
            conn.commit();//手动提交

            XxlJobLogger.log("[发送]开始发送异常流水:" + list.size());
            KafkaProducer producer = KafkaUtil.getKafkaProducer(kafkaBS,kafkaKS,kafkaVS);
            ThreadPoolExecutor tpe = new ThreadPoolExecutor(corePoolSize, maxPoolSize, 0,
                    TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());//,new ThreadPoolExecutor.CallerRunsPolicy());
            List<Future<String>> futureList = new ArrayList<>();
            List<Object[]> failListNos = new ArrayList<Object[]>();
            for(int i = 0; i < list.size(); i++){
                GrantryTransMainErrEntity entity = list.get(i);
                String tradeId = entity.getTradeid();
                String jsonName = entity.getJsonname();//包名
                short seq = entity.getSeq();

                //复制对象到用于发送kafka的对象中
                pojo = new GrantryTransMainErrPojo();
                BeanUtils.copyProperties(entity,pojo);

                //将bean对象转换成json对象用于发送kafka
                if(1 == entity.getHaddeal() || 3 == entity.getHaddeal()) {//只有为 1-已治理 状态时才进行发送,3-车道软件厂家治理
                    String jsonStr = JSON.toJSONString(pojo);
                    jsonStr = JSON.toJSONString(JSONObject.parseObject(jsonStr, GrantryTransMain.class));
                    JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                    JSONArray jsonArray = new JSONArray();
                    jsonArray.add(jsonObject);

                    Task task = new Task(producer,tradeId,jsonArray,getJsonname(jsonName, seq));
                    //调用submit可以获得线程的返回值
                    Future<String> result = tpe.submit(task);
                    futureList.add(result);
                }
            }
            //closeKafkaProducer(producer);

            for(Future<String> result : futureList){
                if(!"success".equals(result.get())) {
                    failListNos.add(new Object[]{result.get()});
                }
            }
            tpe.shutdown();

            //将发送失败记录的haddeal状态更新为 1-已治理
            jdbcTemplate.batchUpdate(getSqlStr(Constants.SQL_UPD_CHARGE_GRANTRY_MAIN),failListNos);
            conn.commit();//手动提交

            XxlJobLogger.log("[普通]已治理状态记录发送失败(条):" + failListNos.size());
            XxlJobLogger.log("=====完成执行业务操作===");
        }catch(Exception e){
            if (conn != null) {
                try {
                    conn.rollback();
                    XxlJobLogger.log("回滚成功.");
                }catch (Exception e1){
                    e1.printStackTrace();
                }
            }
            XxlJobLogger.log("抛出异常："+e.getMessage());
            throw e;
        }finally{
            if (conn != null) {
                try {
                    conn.setAutoCommit(true);
                    conn.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 用httpclient发送数据给中台
     * @param datalist
     * @param jsonName
     * @throws IOException
     */
    public int sendPlatform(JSONArray datalist, String jsonName) throws Exception {
        String url = middlegroundUrl;
        byte[] bytes = JSONArray.toJSONBytes(datalist);

        bytes = compress(bytes);
//        XxlJobLogger.log("bytes长度："+bytes.length);
        String md5 = DigestUtils.md5Hex(bytes);
//        XxlJobLogger.log("md5："+md5);
        InputStream sbs = new ByteArrayInputStream(bytes);

        Map<String, String> headMap = new HashMap<>();
        Map<String, String> bodyMap = new HashMap<>();
        headMap.put("binfile-gzip", null);
        headMap.put("binfile-md5", md5);
        headMap.put("binfile-auth", null);
        bodyMap.put("filename", jsonName);

        HttpResponse response = HttpClientUtils.postSutongFile(url, sbs, jsonName, headMap, bodyMap);
        BufferedReader reader = new BufferedReader(new InputStreamReader(response.getEntity().getContent(), "UTF-8")); // 实例化输入流，并获取网页代码
        String s; // 依次循环，至到读的值为空
        StringBuilder sb = new StringBuilder();
        while ((s = reader.readLine()) != null) {
            sb.append(s);
        }
        reader.close();
        String str = sb.toString();
        int status = response.getStatusLine().getStatusCode();
        JSONObject jsonObject = JSONObject.parseObject(str);
        if (status != 200) {
            XxlJobLogger.log(jsonObject.get("info").toString());
            XxlJobLogger.log("[失败]修复数据发送失败,包名为：" + jsonName);
            //throw new Exception("请求中台http接口失败,响应状态码："+status);
        }
        return status;

    }

    /**
     * 关闭kafka对象
     * @param producer
     */
//    public void closeKafkaProducer(KafkaProducer producer){
//        if(null != producer){
//            producer.close();
//        }
//    }

    /**
     * kafka发送数据
     * @param datalist
     * @param jsonName
     * @throws Exception
     */
    public void sendToKafka(KafkaProducer producer,JSONArray datalist, String jsonName) throws Exception {
        if(null == producer){
            throw new Exception("kafka生产者对象为空,发送失败!");
        }

        String topicName = null;
        String fileType = jsonName.split("_REQ")[0];
        if("BASIC_BINLIST".equalsIgnoreCase(fileType)){
            topicName = "mtcbinlist";
        }else if("BASIC_BOUTLIST".equalsIgnoreCase(fileType)){
            topicName = "mtcboutlist";
        }else if("BASIC_ETCBINLIST".equalsIgnoreCase(fileType)){
            topicName = "etcbinlist";
        }else if("BASIC_ETCBOUTLIST".equalsIgnoreCase(fileType)){
            topicName = "etcboutlist";
        }else if("GBUPLOAD_ETCTU".equalsIgnoreCase(fileType) || "BASIC_GRANTRYTRANSMAIN".equalsIgnoreCase(fileType)){
            topicName = "grantrytransmain";
        }

        String list = datalist.toString();
        ProducerRecord record = new ProducerRecord(topicName, jsonName, list);
        Future result = null;
        try {
            result = producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e != null) {
                        XxlJobLogger.log("异常");
                        throw new BusinessException(StatusCodeEnum.SYS_OPR_ERR,"异常");
                    }
                }
            });
        } catch (Exception e) {
            XxlJobLogger.log("[失败]修复数据发送失败,包名为:" + jsonName);
        }
        XxlJobLogger.log(String.valueOf(result.get()));
    }

    /**
     * 获取文件名
     * @param jsonname
     * @param seq
     * @return
     */
    private String getJsonname(String jsonname,short seq){
        if(!StringUtils.isEmpty(jsonname)){
            String dateStr = jsonname.substring(jsonname.lastIndexOf("_"));
            if(dateStr.length() > 17){
                jsonname = jsonname.substring(0,jsonname.lastIndexOf("_")).concat(dateStr.substring(0,18).concat(String.valueOf(seq)));
            }else{
                jsonname = jsonname.concat(String.valueOf(seq));
            }

//            if(!jsonname.contains(".")){
//                //由于数据已在前面进行了压缩,故发送中台时，没有带后缀的包名后直接拼接一个.zip后缀
//                jsonname = jsonname.concat(".zip");
//            }
            return jsonname;
        }else{
            return null;
        }
    }

    /**
     * 根据sql标识获取相应的sql语句
     * @param sqlId
     * @return
     */
    private String getSqlStr(String sqlId){
        StringBuffer sql = new StringBuffer();
        if(Constants.SQL_INST_CHARGE_BINLIST.equals(sqlId)) {//MTC入口流水表
            sql.append("insert into tb_charge_binlist(VEHICLESIGNID,ENTOLLLANEHEX,PASSID,RECEIVETIME,PACKNAME,TRANSFEE," +
                    "       UPDATESTATE,ENSTATIONHEX,VEHICLESIGN,INSERTTABLETIME,SEQ,ERRORTYPE,ERRORDETAIL,HADDEAL,DEALTIME," +
                    "       ERRORTOTAL,JSONNAME,LISTNO,RECORDNO,STATIONNO,SQUARENO,LANENO,LANETYPE,SAMID,PROGRAMVER,PVEHTYPE,\n" +
                    "       PVEHCLASS,VEHTYPE,VEHCLASS,VEHPLATE,VEHTYPEAUTO,VEHPLATEAUTO,SQUADNO,SQUADDATE,OPERATORNO,OPERATORNAME,\n" +
                    "       OPTIME,CARDBOXNO,CARDBOXCARDNO,CARDTRUNKNO,TICKETTYPE,PAPERNO,CARDID,CARDNO,PASSNUM,DIRECTION,HIGHSPEVENT,\n" +
                    "       MIDSPEVENT,LOWSPEVENT,SPVEHKIND,KEYPRESSINFO,DEVICESTATUS,ELEOBUID,ELECARDTYPE,ELECARDID,ELETRADENO,ELEPSAMID,\n" +
                    "       ELEPSAMTRADENO,ELETRADETYPE,ELEBALANCE,ELEREBATE,ELETAC,VEHNUM,RECORDTYPE,UNDOLISTNO,RECORDTTAG,CARDMARK,\n" +
                    "       AXISINFO,AXISTYPE,AXISNUM,TOTALWEIGHT,LIMITWEIGHT,IDENTIFYSTATUS,GOODSTYPE,SPARE1,SPARE2,SPARE3,SPARE4,\n" +
                    "       SPARE5,SPARE6,VERIFYCODE,TRANSFERTAG,MEDIATYPE,OBUSIGN,TRANSTYPE,ELECTRICALPERCENTAGE,SIGNSTATUS,\n" +
                    "       DESCRIPTION,SPECIALTYPE,updateuser,dealdesc,createTime)\n" +
                    "SELECT VEHICLESIGNID,ENTOLLLANEHEX,PASSID,RECEIVETIME,PACKNAME,TRANSFEE,\n" +
                    "       UPDATESTATE,ENSTATIONHEX,VEHICLESIGN,INSERTTABLETIME,SEQ,ERRORTYPE,ERRORDETAIL,decode(haddeal,1,2,haddeal) as haddeal,DEALTIME,\n" +
                    "       ERRORTOTAL,JSONNAME,LISTNO,RECORDNO,STATIONNO,SQUARENO,LANENO,LANETYPE,SAMID,PROGRAMVER,PVEHTYPE,\n" +
                    "       PVEHCLASS,VEHTYPE,VEHCLASS,VEHPLATE,VEHTYPEAUTO,VEHPLATEAUTO,SQUADNO,SQUADDATE,OPERATORNO,OPERATORNAME,\n" +
                    "       OPTIME,CARDBOXNO,CARDBOXCARDNO,CARDTRUNKNO,TICKETTYPE,PAPERNO,CARDID,CARDNO,PASSNUM,DIRECTION,HIGHSPEVENT,\n" +
                    "       MIDSPEVENT,LOWSPEVENT,SPVEHKIND,KEYPRESSINFO,DEVICESTATUS,ELEOBUID,ELECARDTYPE,ELECARDID,ELETRADENO,ELEPSAMID,\n" +
                    "       ELEPSAMTRADENO,ELETRADETYPE,ELEBALANCE,ELEREBATE,ELETAC,VEHNUM,RECORDTYPE,UNDOLISTNO,RECORDTTAG,CARDMARK,\n" +
                    "       AXISINFO,AXISTYPE,AXISNUM,TOTALWEIGHT,LIMITWEIGHT,IDENTIFYSTATUS,GOODSTYPE,SPARE1,SPARE2,SPARE3,SPARE4,\n" +
                    "       SPARE5,SPARE6,VERIFYCODE,TRANSFERTAG,MEDIATYPE,OBUSIGN,TRANSTYPE,ELECTRICALPERCENTAGE,SIGNSTATUS,\n" +
                    "       DESCRIPTION,SPECIALTYPE,updateuser,dealdesc,sysdate\n" +
                    "FROM binlist_err WHERE haddeal in(1,3,4) and listno = ?");
        }else if(Constants.SQL_INST_CHARGE_ETCBINLIST.equals(sqlId)){//ETC入口流水表
            sql.append("insert into tb_charge_etcbinlist(LISTNO,RECORDNO,STATIONNO,SQUARENO,LANENO,LANETYPE,SAMID,PROGRAMVER,PVEHTYPE,\n" +
                    "       PVEHCLASS,VEHTYPE,VEHCLASS,VEHPLATE,VEHTYPEAUTO,VEHPLATEAUTO,SQUADNO,SQUADDATE,OPERATORNO,OPERATORNAME,\n" +
                    "       OPTIME,CARDBOXNO,CARDBOXCARDNO,CARDTRUNKNO,TICKETTYPE,PAPERNO,CARDID,CARDNO,PASSNUM,DIRECTION,HIGHSPEVENT,\n" +
                    "       MIDSPEVENT,LOWSPEVENT,SPVEHKIND,KEYPRESSINFO,DEVICESTATUS,ELEOBUID,ELECARDTYPE,ELECARDID,ELETRADENO,\n" +
                    "       ELEPSAMID,ELEPSAMTRADENO,ELETRADETYPE,ELEBALANCE,ELEREBATE,ELETAC,VEHNUM,RECORDTYPE,UNDOLISTNO,\n" +
                    "       RECORDTTAG,CARDMARK,AXISINFO,AXISTYPE,AXISNUM,TOTALWEIGHT,LIMITWEIGHT,IDENTIFYSTATUS,GOODSTYPE,\n" +
                    "       SPARE1,SPARE2,SPARE3,SPARE4,SPARE5,SPARE6,VERIFYCODE,TRANSFERTAG,MEDIATYPE,OBUSIGN,TRANSTYPE,\n" +
                    "       ELECTRICALPERCENTAGE,SIGNSTATUS,DESCRIPTION,SPECIALTYPE,VEHICLESIGNID,ENTOLLLANEHEX,PASSID,RECEIVETIME,\n" +
                    "       PACKNAME,TRANSFEE,UPDATESTATE,ENSTATIONHEX,VEHICLESIGN,INSERTTABLETIME,SEQ,ERRORTYPE,ERRORDETAIL,\n" +
                    "       HADDEAL,DEALTIME,ERRORTOTAL,JSONNAME,updateuser,dealdesc,createTime)\n" +
                    "SELECT LISTNO,RECORDNO,STATIONNO,SQUARENO,LANENO,LANETYPE,SAMID,PROGRAMVER,PVEHTYPE,\n" +
                    "       PVEHCLASS,VEHTYPE,VEHCLASS,VEHPLATE,VEHTYPEAUTO,VEHPLATEAUTO,SQUADNO,SQUADDATE,OPERATORNO,OPERATORNAME,\n" +
                    "       OPTIME,CARDBOXNO,CARDBOXCARDNO,CARDTRUNKNO,TICKETTYPE,PAPERNO,CARDID,CARDNO,PASSNUM,DIRECTION,HIGHSPEVENT,\n" +
                    "       MIDSPEVENT,LOWSPEVENT,SPVEHKIND,KEYPRESSINFO,DEVICESTATUS,ELEOBUID,ELECARDTYPE,ELECARDID,ELETRADENO,\n" +
                    "       ELEPSAMID,ELEPSAMTRADENO,ELETRADETYPE,ELEBALANCE,ELEREBATE,ELETAC,VEHNUM,RECORDTYPE,UNDOLISTNO,\n" +
                    "       RECORDTTAG,CARDMARK,AXISINFO,AXISTYPE,AXISNUM,TOTALWEIGHT,LIMITWEIGHT,IDENTIFYSTATUS,GOODSTYPE,\n" +
                    "       SPARE1,SPARE2,SPARE3,SPARE4,SPARE5,SPARE6,VERIFYCODE,TRANSFERTAG,MEDIATYPE,OBUSIGN,TRANSTYPE,\n" +
                    "       ELECTRICALPERCENTAGE,SIGNSTATUS,DESCRIPTION,SPECIALTYPE,VEHICLESIGNID,ENTOLLLANEHEX,PASSID,RECEIVETIME,\n" +
                    "       PACKNAME,TRANSFEE,UPDATESTATE,ENSTATIONHEX,VEHICLESIGN,INSERTTABLETIME,SEQ,ERRORTYPE,ERRORDETAIL,\n" +
                    "       decode(haddeal,1,2,haddeal) as haddeal,DEALTIME,ERRORTOTAL,JSONNAME,updateuser,dealdesc,sysdate\n" +
                    "FROM etcbinlist_err WHERE haddeal in(1,3,4) and listno = ?");
        }else if(Constants.SQL_INST_CHARGE_BOUTLIST.equals(sqlId)){//MTC出口流水表
            sql.append("insert into tb_charge_boutlist(VERIFYCODE,VEHTYPEAUTO,VEHTYPE,VEHPLATEAUTO,VEHPLATE,VEHNUM,VEHICLESIGNID,\n" +
                    "       VEHICLESIGN,VEHCLASS,UPDATESTATE,UNDOLISTNO,TRANSTYPE,TRANSFERTAG,TRANSFEE,TOTALWEIGHT,TOTALTOLL,\n" +
                    "       TOTALCOUNT,TOLLSUPPORT,TOLLPROVINCEID,TICKETTYPE,STATIONNO,SQUARENO,SQUADNO,SQUADDATE,SPVEHKIND,\n" +
                    "       SPLITPROVINCE,SPECIALTYPE,SPCRATEVERSION,SPARE6,SPARE5,SPARE4,SPARE3,SPARE2,SPARE1,SIGNSTATUS,\n" +
                    "       SHORTFEEMILEAGE,SHORTFEE,SERVICETYPE,SEQ,SAMID,RECORDTYPE,RECORDTTAG,RECORDNO,RECEIVETIME,REBATEVER,\n" +
                    "       RATEVERSION,PVEHTYPE,PVEHCLASS,PTOTALTOLL,PROVTRANSCOUNT,PROVINCEDISCOUNTFEE,PROVINCECOUNT,PROGRAMVER,\n" +
                    "       PRICEVERNO,PAYTYPE,PAXISINFO,PASSNUM,PASSID,PACKNAME,OVERLOADRATE,ORIGINFEE,OPTIME,OPERATORNO,OPERATORNAME,\n" +
                    "       OBUTOTALDISCOUNTAMOUNT,OBUTOTALAMOUNT,OBUSIGN,NOCARDCOUNT,MULTIPROVINCE,MIDSPEVENT,MEDIATYPE,LOWSPEVENT,\n" +
                    "       LISTNO,LINE_SPECIALTYPE,LINE_SPECIALDESC,LINE_GANTRY_INTERVAL,LINE_FACT_CACLTYPE,LINE_ENDHEX_SHORT,\n" +
                    "       LINE_ENDHEX,LINE_CPC_TOTALFEE,LINE_CPC_PASSINTERVAL,LINE_CPC_PASSFEE,LINE_CPC_LASTGANTRYHEX,LINE_CALCSTATE_FINAL,\n" +
                    "       LINE_CALCSTATE,LINE_BEGINHEX_SHORT,LINE_BEGINHEX,LIMITWEIGHT,LASTSPEVENT,LANETYPE,LANENO,KEYPRESSINFO,\n" +
                    "       JSONNAME,INVOICETYPE,INVOICEID,INVEHTYPE,INVEHPLATE,INVEHCLASS,INTOTALWEIGHT,INSTATIONNO,INSQUARENO,\n" +
                    "       INSERTTABLETIME,INSAMNO,INPSAMNO,INOPTIME,INOPERATORNO,INLISTNO,INLANENO,INDIRECTION,INAXISTYPE,IDENTIFYSTATUS,\n" +
                    "       IDENTIFICATION,HIGHSPEVENT,HADDEAL,GOODSTYPE,FREEREGION,FLAGSTATIONID,FEERATE,FEEMILEAGE,FEE,EXTOLLSTATIONNAME,\n" +
                    "       EXTOLLLANEID,EXSTATIONHEX,EXITFEETYPE,ETCTOTALAMOUNT,ERRORTYPE,ERRORTOTAL,ERRORDETAIL,ENWEIGHT,ENTRYTOTALTOLL,\n" +
                    "       ENTOLLSTATIONNAME,ENTOLLLANEID,ENTOLLLANEHEX,ENSTATIONHEX,ENCRYPVER,ENAXLECOUNT,ELETRADETYPE,ELETRADETAC,\n" +
                    "       ELETRADENO,ELETERMID,ELETAC,ELEREBATE,ELEPSAMTRADENO,ELEPSAMID,ELEPAYMETHOD,ELEOBUID,ELELISTNO,ELECTRICALPERCENTAGE,\n" +
                    "       ELECARDTYPE,ELECARDNO,ELECARDID,ELEBBALANCE,ELEABALANCE,EDECRERECORD,DISCOUNTTYPE,DISCOUNT,DIRECTION,\n" +
                    "       DEVICESTATUS,DESCRIPTION,DEALTIME,DATATYPE,CARDTRUNKNO,CARDNO,CARDMARK,CARDID,CARDBOXNO,CARDBOXCARDNO,\n" +
                    "       BVCLASS,BANKTYPE,BANKTRXNO,BANKTRANTAC,BANKTRADENO,BANKTERMID,BANKERRORNO,AXISTYPE,AXISNUM,AXISINFO,\n" +
                    "       AXISEXCEEDRATE,AXISEDITION,ALGORITHMIDENTIFIER,updateuser,dealdesc,createTime)\n" +
                    "SELECT VERIFYCODE,VEHTYPEAUTO,VEHTYPE,VEHPLATEAUTO,VEHPLATE,VEHNUM,VEHICLESIGNID,\n" +
                    "       VEHICLESIGN,VEHCLASS,UPDATESTATE,UNDOLISTNO,TRANSTYPE,TRANSFERTAG,TRANSFEE,TOTALWEIGHT,TOTALTOLL,\n" +
                    "       TOTALCOUNT,TOLLSUPPORT,TOLLPROVINCEID,TICKETTYPE,STATIONNO,SQUARENO,SQUADNO,SQUADDATE,SPVEHKIND,\n" +
                    "       SPLITPROVINCE,SPECIALTYPE,SPCRATEVERSION,SPARE6,SPARE5,SPARE4,SPARE3,SPARE2,SPARE1,SIGNSTATUS,\n" +
                    "       SHORTFEEMILEAGE,SHORTFEE,SERVICETYPE,SEQ,SAMID,RECORDTYPE,RECORDTTAG,RECORDNO,RECEIVETIME,REBATEVER,\n" +
                    "       RATEVERSION,PVEHTYPE,PVEHCLASS,PTOTALTOLL,PROVTRANSCOUNT,PROVINCEDISCOUNTFEE,PROVINCECOUNT,PROGRAMVER,\n" +
                    "       PRICEVERNO,PAYTYPE,PAXISINFO,PASSNUM,PASSID,PACKNAME,OVERLOADRATE,ORIGINFEE,OPTIME,OPERATORNO,OPERATORNAME,\n" +
                    "       OBUTOTALDISCOUNTAMOUNT,OBUTOTALAMOUNT,OBUSIGN,NOCARDCOUNT,MULTIPROVINCE,MIDSPEVENT,MEDIATYPE,LOWSPEVENT,\n" +
                    "       LISTNO,LINE_SPECIALTYPE,LINE_SPECIALDESC,LINE_GANTRY_INTERVAL,LINE_FACT_CACLTYPE,LINE_ENDHEX_SHORT,\n" +
                    "       LINE_ENDHEX,LINE_CPC_TOTALFEE,LINE_CPC_PASSINTERVAL,LINE_CPC_PASSFEE,LINE_CPC_LASTGANTRYHEX,LINE_CALCSTATE_FINAL,\n" +
                    "       LINE_CALCSTATE,LINE_BEGINHEX_SHORT,LINE_BEGINHEX,LIMITWEIGHT,LASTSPEVENT,LANETYPE,LANENO,KEYPRESSINFO,\n" +
                    "       JSONNAME,INVOICETYPE,INVOICEID,INVEHTYPE,INVEHPLATE,INVEHCLASS,INTOTALWEIGHT,INSTATIONNO,INSQUARENO,\n" +
                    "       INSERTTABLETIME,INSAMNO,INPSAMNO,INOPTIME,INOPERATORNO,INLISTNO,INLANENO,INDIRECTION,INAXISTYPE,IDENTIFYSTATUS,\n" +
                    "       IDENTIFICATION,HIGHSPEVENT,decode(haddeal,1,2,haddeal) as haddeal,GOODSTYPE,FREEREGION,FLAGSTATIONID,FEERATE,FEEMILEAGE,FEE,EXTOLLSTATIONNAME,\n" +
                    "       EXTOLLLANEID,EXSTATIONHEX,EXITFEETYPE,ETCTOTALAMOUNT,ERRORTYPE,ERRORTOTAL,ERRORDETAIL,ENWEIGHT,ENTRYTOTALTOLL,\n" +
                    "       ENTOLLSTATIONNAME,ENTOLLLANEID,ENTOLLLANEHEX,ENSTATIONHEX,ENCRYPVER,ENAXLECOUNT,ELETRADETYPE,ELETRADETAC,\n" +
                    "       ELETRADENO,ELETERMID,ELETAC,ELEREBATE,ELEPSAMTRADENO,ELEPSAMID,ELEPAYMETHOD,ELEOBUID,ELELISTNO,ELECTRICALPERCENTAGE,\n" +
                    "       ELECARDTYPE,ELECARDNO,ELECARDID,ELEBBALANCE,ELEABALANCE,EDECRERECORD,DISCOUNTTYPE,DISCOUNT,DIRECTION,\n" +
                    "       DEVICESTATUS,DESCRIPTION,DEALTIME,DATATYPE,CARDTRUNKNO,CARDNO,CARDMARK,CARDID,CARDBOXNO,CARDBOXCARDNO,\n" +
                    "       BVCLASS,BANKTYPE,BANKTRXNO,BANKTRANTAC,BANKTRADENO,BANKTERMID,BANKERRORNO,AXISTYPE,AXISNUM,AXISINFO,\n" +
                    "       AXISEXCEEDRATE,AXISEDITION,ALGORITHMIDENTIFIER,updateuser,dealdesc,sysdate\n" +
                    "FROM boutlist_err WHERE haddeal in(1,3,4) and listno = ?");
        }else if(Constants.SQL_INST_CHARGE_ETCBOUTLIST.equals(sqlId)){//ETC出口流水表
            sql.append("insert into tb_charge_etcboutlist(PACKNAME,UPDATESTATE,VEHICLESIGN,FEE,NOCARDCOUNT,SHORTFEE,FEERATE,EXITFEETYPE,\n" +
                    "       SPCRATEVERSION,PROVFEE,PROVTRANSCOUNT,PROVINCECOUNT,DISCOUNTTYPE,PROVINCEDISCOUNTFEE,ORIGINFEE,FEEMILEAGE,\n" +
                    "       TOLLPROVINCEID,ISSUERID,OBUTOTALAMOUNT,OBUTOTALDISCOUNTAMOUNT,ETCTOTALAMOUNT,SHORTFEEMILEAGE,FEEBOARDPLAY,\n" +
                    "       OBUTOTALCOUNT,ENSTATIONHEX,INSERTTABLETIME,SEQ,ERRORTYPE,ERRORDETAIL,HADDEAL,DEALTIME,ERRORTOTAL,JSONNAME,\n" +
                    "       RECORDTYPE,UNDOLISTNO,RECORDTTAG,CARDMARK,ENCRYPVER,IDENTIFYSTATUS,ELELISTNO,ELEOBUID,ELECARDTYPE,\n" +
                    "       ELECARDNO,ELECARDID,ELETRADENO,ELEPSAMID,ELEPSAMTRADENO,ELETRADETYPE,ELEPAYMETHOD,ELEBBALANCE,ELEABALANCE,\n" +
                    "       ELEREBATE,ELETAC,BANKTYPE,BANKERRORNO,BANKTRADENO,BANKTRXNO,BANKTRANTAC,BANKTERMID,ELETRADETAC,ELETERMID,\n" +
                    "       EDECRERECORD,SPARE1,SPARE2,SPARE3,SPARE4,SPARE5,SPARE6,VERIFYCODE,TRANSFERTAG,MEDIATYPE,TRANSTYPE,ENTOLLLANEID,\n" +
                    "       ENTOLLLANEHEX,EXTOLLLANEID,ENTOLLSTATIONNAME,EXTOLLSTATIONNAME,SERVICETYPE,ALGORITHMIDENTIFIER,DISCOUNT,PAYTYPE,\n" +
                    "       IDENTIFICATION,DESCRIPTION,SPECIALTYPE,VEHICLESIGNID,PASSID,TOLLSUPPORT,MULTIPROVINCE,OBUSIGN,\n" +
                    "       ELECTRICALPERCENTAGE,SIGNSTATUS,DATATYPE,SPLITPROVINCE,TRANSFEE,ENAXLECOUNT,ENWEIGHT,RECEIVETIME,\n" +
                    "       LISTNO,RECORDNO,STATIONNO,SQUARENO,LANENO,LANETYPE,SAMID,PROGRAMVER,INLISTNO,INSTATIONNO,INLANENO,INVEHTYPE,\n" +
                    "       INVEHCLASS,INVEHPLATE,INOPERATORNO,INOPTIME,INSQUARENO,INDIRECTION,INAXISTYPE,INTOTALWEIGHT,INPSAMNO,INSAMNO,\n" +
                    "       PVEHTYPE,PVEHCLASS,BVCLASS,VEHTYPE,VEHCLASS,VEHPLATE,VEHTYPEAUTO,VEHPLATEAUTO,SQUADNO,SQUADDATE,OPERATORNO,\n" +
                    "       OPERATORNAME,OPTIME,CARDBOXNO,CARDBOXCARDNO,CARDTRUNKNO,TICKETTYPE,CARDID,CARDNO,PASSNUM,DIRECTION,\n" +
                    "       HIGHSPEVENT,MIDSPEVENT,LOWSPEVENT,LASTSPEVENT,SPVEHKIND,ENTRYTOTALTOLL,PTOTALTOLL,TOTALTOLL,INVOICEID,\n" +
                    "       INVOICETYPE,AXISNUM,TOTALWEIGHT,LIMITWEIGHT,OVERLOADRATE,AXISTYPE,PAXISINFO,AXISINFO,AXISEXCEEDRATE,\n" +
                    "       PRICEVERNO,AXISEDITION,GOODSTYPE,REBATEVER,FLAGSTATIONID,FREEREGION,KEYPRESSINFO,DEVICESTATUS,VEHNUM,updateuser,dealdesc,createTime)\n" +
                    "SELECT PACKNAME,UPDATESTATE,VEHICLESIGN,FEE,NOCARDCOUNT,SHORTFEE,FEERATE,EXITFEETYPE,\n" +
                    "       SPCRATEVERSION,PROVFEE,PROVTRANSCOUNT,PROVINCECOUNT,DISCOUNTTYPE,PROVINCEDISCOUNTFEE,ORIGINFEE,FEEMILEAGE,\n" +
                    "       TOLLPROVINCEID,ISSUERID,OBUTOTALAMOUNT,OBUTOTALDISCOUNTAMOUNT,ETCTOTALAMOUNT,SHORTFEEMILEAGE,FEEBOARDPLAY,\n" +
                    "       OBUTOTALCOUNT,ENSTATIONHEX,INSERTTABLETIME,SEQ,ERRORTYPE,ERRORDETAIL,decode(haddeal,1,2,haddeal) as haddeal,DEALTIME,ERRORTOTAL,JSONNAME,\n" +
                    "       RECORDTYPE,UNDOLISTNO,RECORDTTAG,CARDMARK,ENCRYPVER,IDENTIFYSTATUS,ELELISTNO,ELEOBUID,ELECARDTYPE,\n" +
                    "       ELECARDNO,ELECARDID,ELETRADENO,ELEPSAMID,ELEPSAMTRADENO,ELETRADETYPE,ELEPAYMETHOD,ELEBBALANCE,ELEABALANCE,\n" +
                    "       ELEREBATE,ELETAC,BANKTYPE,BANKERRORNO,BANKTRADENO,BANKTRXNO,BANKTRANTAC,BANKTERMID,ELETRADETAC,ELETERMID,\n" +
                    "       EDECRERECORD,SPARE1,SPARE2,SPARE3,SPARE4,SPARE5,SPARE6,VERIFYCODE,TRANSFERTAG,MEDIATYPE,TRANSTYPE,ENTOLLLANEID,\n" +
                    "       ENTOLLLANEHEX,EXTOLLLANEID,ENTOLLSTATIONNAME,EXTOLLSTATIONNAME,SERVICETYPE,ALGORITHMIDENTIFIER,DISCOUNT,PAYTYPE,\n" +
                    "       IDENTIFICATION,DESCRIPTION,SPECIALTYPE,VEHICLESIGNID,PASSID,TOLLSUPPORT,MULTIPROVINCE,OBUSIGN,\n" +
                    "       ELECTRICALPERCENTAGE,SIGNSTATUS,DATATYPE,SPLITPROVINCE,TRANSFEE,ENAXLECOUNT,ENWEIGHT,RECEIVETIME,\n" +
                    "       LISTNO,RECORDNO,STATIONNO,SQUARENO,LANENO,LANETYPE,SAMID,PROGRAMVER,INLISTNO,INSTATIONNO,INLANENO,INVEHTYPE,\n" +
                    "       INVEHCLASS,INVEHPLATE,INOPERATORNO,INOPTIME,INSQUARENO,INDIRECTION,INAXISTYPE,INTOTALWEIGHT,INPSAMNO,INSAMNO,\n" +
                    "       PVEHTYPE,PVEHCLASS,BVCLASS,VEHTYPE,VEHCLASS,VEHPLATE,VEHTYPEAUTO,VEHPLATEAUTO,SQUADNO,SQUADDATE,OPERATORNO,\n" +
                    "       OPERATORNAME,OPTIME,CARDBOXNO,CARDBOXCARDNO,CARDTRUNKNO,TICKETTYPE,CARDID,CARDNO,PASSNUM,DIRECTION,\n" +
                    "       HIGHSPEVENT,MIDSPEVENT,LOWSPEVENT,LASTSPEVENT,SPVEHKIND,ENTRYTOTALTOLL,PTOTALTOLL,TOTALTOLL,INVOICEID,\n" +
                    "       INVOICETYPE,AXISNUM,TOTALWEIGHT,LIMITWEIGHT,OVERLOADRATE,AXISTYPE,PAXISINFO,AXISINFO,AXISEXCEEDRATE,\n" +
                    "       PRICEVERNO,AXISEDITION,GOODSTYPE,REBATEVER,FLAGSTATIONID,FREEREGION,KEYPRESSINFO,DEVICESTATUS,VEHNUM,updateuser,dealdesc,sysdate\n" +
                    "FROM etcboutlist_err WHERE haddeal in(1,3,4) and listno = ?");
        }else if(Constants.SQL_INST_CHARGE_GRANTRY_MAIN.equals(sqlId)){//门架流水主表
            sql.append("insert into tb_charge_grantry_trans_main(RATEFITCOUNT,OBUFEESUMBEFORE,OBUFEESUMAFTER,OBUPROVFEESUMBEFORE,\n" +
                    "       OBUPROVFEESUMAFTER,CARDFEESUMBEFORE,CARDFEESUMAFTER,NOCARDTIMESBEFORE,NOCARDTIMESAFTER,PROVINCENUMBEFORE,\n" +
                    "       PROVINCENUMAFTER,OBUTOTALTRADESUCCNUMBEFORE,OBUTOTALTRADESUCCNUMAFTER,OBUPROVTRADESUCCNUMBEFORE,\n" +
                    "       OBUPROVTRADESUCCNUMAFTER,OBUTRADERESULT,TRADETYPE,OBUINFOTYPEREAD,OBUINFOTYPEWRITE,OBUPASSSTATE,\n" +
                    "       FEEVEHICLETYPE,OBULASTGANTRYHEX,OBULASTGANTRYTIME,CHARGESSPECIALTYPE,OBUVERIFYCODE,TRANSACTIONTYPE,\n" +
                    "       VEHICLESIGN,FEECALCSPECIAL,LASTGANTRYHEXFEE,LASTGANTRYHEXPASS,ISFIXDATA,PROVMINFEE,FEESPARE1,FEESPARE2,\n" +
                    "       FEESPARE3,FEEPROVBEGINHEX,OBUPAYFEESUMBEFORE,OBUPAYFEESUMAFTER,OBUDISCOUNTFEESUMBEFORE,\n" +
                    "       OBUDISCOUNTFEESUMAFTER,FEEMILEAGE,OBUMILEAGEBEFORE,OBUMILEAGEAFTER,GANTRYTYPE,HGSENDFLAG,CPCFEETRADERESULT,\n" +
                    "       FEECALCSPECIALS,FEEPROVBEGINHEXFIT,FEEPROVBEGINTIMEFIT,FEEPROVEF04,FITPROVFLAG,GANTRYPASSCOUNTBEFORE,\n" +
                    "       GANTRYPASSINFOAFTER,OBUPROVPAYFEESUMAFTER,OBUPROVPAYFEESUMBEFORE,PATHFITFLAG,PAYFEEPROVSUMLOCAL,\n" +
                    "       PCRSUVERSION,READCIPHERTEXTVERIFY,TRADEREADCIPHERTEXT,TRADEWRITECIPHERTEXT,UPDATERESULT,\n" +
                    "       TOLLINTERVALSIGN,PROVMINFEECALCMODE,INSERTTABLETIME,FEEPROVBEGINTIME,FEESUMLOCALAFTEREF04,\n" +
                    "       LASTGANTRYFEEPASS,LASTGANTRYMILEPASS,SEQ,ERRORTYPE,ERRORDETAIL,HADDEAL,DEALTIME,ERRORTOTAL,JSONNAME,\n" +
                    "       TRADEID,SECTIONID,GANTRYID,COMPUTERORDER,HOURBATCHNO,GANTRYORDERNUM,GANTRYHEX,GANTRYHEXOPPOSITE,\n" +
                    "       TRANSTIME,PAYFEE,FEE,DISCOUNTFEE,TRANSFEE,MEDIATYPE,OBUSIGN,ETCCARDTYPE,TERMINALTRANSNO,TOLLINTERVALID,\n" +
                    "       TOLLINTERVALNAME,FEEGROUP,PAYFEEGROUP,DISCOUNTFEEGROUP,VEHICLEPLATE,VEHICLETYPE,IDENTIFYVEHICLETYPE,\n" +
                    "       VEHICLECLASS,TAC,TRANSTYPE,TERMINALNO,SERVICETYPE,ALGORITHMIDENTIFIER,RATEVERSION,KEYVERSION,ANTENNAID,\n" +
                    "       OBUELECTRICAL,CONSUMETIME,PASSSTATE,INTERRUPTSIGNAL,SPECIALTYPE,SIGNSTATUS,VEHICLESIGNID,ENTOLLLANEID,\n" +
                    "       ENTOLLSTATIONHEX,ENTIME,ENLANETYPE,PASSID,LASTGANTRYHEX,LASTGANTRYTIME,OBUMAC,OBUISSUEID,OBUSN,OBUVERSION,\n" +
                    "       OBUSTARTDATE,OBUENDDATE,OBUSTATE,OBUVEHICLEPLATE,OBUVEHICLETYPE,VEHICLEUSERTYPE,VEHICLESEAT,AXLECOUNT,\n" +
                    "       TOTALWEIGHT,VEHICLELENGTH,VEHICLEWIDTH,VEHICLEHIGHT,CPUNETID,CPUISSUEID,CPUVEHICLEPLATE,CPUVEHICLETYPE,\n" +
                    "       CPUSTARTDATE,CPUENDDATE,CPUVERSION,CPUCARDTYPE,CPUCARDID,BALANCEBEFORE,BALANCEAFTER,GANTRYPASSCOUNT,\n" +
                    "       GANTRYPASSINFO,FEEPROVINFO,FEESUMLOCALBEFORE,FEESUMLOCALAFTER,FEECALCRESULT,FEEINFO1,FEEINFO2,\n" +
                    "       FEEINFO3,HOLIDAYSTATE,TRADERESULT,VERIFYCODE,VEHICLEPICID,VEHICLETAILPICID,MATCHSTATUS,VALIDSTATUS,\n" +
                    "       DEALSTATUS,RELATEDTRADEID,ALLRELATEDTRADEID,STATIONDBTIME,STATIONDEALTIME,STATIONVALIDTIME,\n" +
                    "       STATIONMATCHTIME,ENTOLLSTATIONNAME,DESCRIPTION,RECEIVETIME,PACKNAME,TRANSNO,RATECOMPUTE,updateuser,dealdesc,createTime)\n" +
                    "SELECT RATEFITCOUNT,OBUFEESUMBEFORE,OBUFEESUMAFTER,OBUPROVFEESUMBEFORE,\n" +
                    "       OBUPROVFEESUMAFTER,CARDFEESUMBEFORE,CARDFEESUMAFTER,NOCARDTIMESBEFORE,NOCARDTIMESAFTER,PROVINCENUMBEFORE,\n" +
                    "       PROVINCENUMAFTER,OBUTOTALTRADESUCCNUMBEFORE,OBUTOTALTRADESUCCNUMAFTER,OBUPROVTRADESUCCNUMBEFORE,\n" +
                    "       OBUPROVTRADESUCCNUMAFTER,OBUTRADERESULT,TRADETYPE,OBUINFOTYPEREAD,OBUINFOTYPEWRITE,OBUPASSSTATE,\n" +
                    "       FEEVEHICLETYPE,OBULASTGANTRYHEX,OBULASTGANTRYTIME,CHARGESSPECIALTYPE,OBUVERIFYCODE,TRANSACTIONTYPE,\n" +
                    "       VEHICLESIGN,FEECALCSPECIAL,LASTGANTRYHEXFEE,LASTGANTRYHEXPASS,ISFIXDATA,PROVMINFEE,FEESPARE1,FEESPARE2,\n" +
                    "       FEESPARE3,FEEPROVBEGINHEX,OBUPAYFEESUMBEFORE,OBUPAYFEESUMAFTER,OBUDISCOUNTFEESUMBEFORE,\n" +
                    "       OBUDISCOUNTFEESUMAFTER,FEEMILEAGE,OBUMILEAGEBEFORE,OBUMILEAGEAFTER,GANTRYTYPE,HGSENDFLAG,CPCFEETRADERESULT,\n" +
                    "       FEECALCSPECIALS,FEEPROVBEGINHEXFIT,FEEPROVBEGINTIMEFIT,FEEPROVEF04,FITPROVFLAG,GANTRYPASSCOUNTBEFORE,\n" +
                    "       GANTRYPASSINFOAFTER,OBUPROVPAYFEESUMAFTER,OBUPROVPAYFEESUMBEFORE,PATHFITFLAG,PAYFEEPROVSUMLOCAL,\n" +
                    "       PCRSUVERSION,READCIPHERTEXTVERIFY,TRADEREADCIPHERTEXT,TRADEWRITECIPHERTEXT,UPDATERESULT,\n" +
                    "       TOLLINTERVALSIGN,PROVMINFEECALCMODE,INSERTTABLETIME,FEEPROVBEGINTIME,FEESUMLOCALAFTEREF04,\n" +
                    "       LASTGANTRYFEEPASS,LASTGANTRYMILEPASS,SEQ,ERRORTYPE,ERRORDETAIL,decode(haddeal,1,2,haddeal) as haddeal,DEALTIME,ERRORTOTAL,JSONNAME,\n" +
                    "       TRADEID,SECTIONID,GANTRYID,COMPUTERORDER,HOURBATCHNO,GANTRYORDERNUM,GANTRYHEX,GANTRYHEXOPPOSITE,\n" +
                    "       TRANSTIME,PAYFEE,FEE,DISCOUNTFEE,TRANSFEE,MEDIATYPE,OBUSIGN,ETCCARDTYPE,TERMINALTRANSNO,TOLLINTERVALID,\n" +
                    "       TOLLINTERVALNAME,FEEGROUP,PAYFEEGROUP,DISCOUNTFEEGROUP,VEHICLEPLATE,VEHICLETYPE,IDENTIFYVEHICLETYPE,\n" +
                    "       VEHICLECLASS,TAC,TRANSTYPE,TERMINALNO,SERVICETYPE,ALGORITHMIDENTIFIER,RATEVERSION,KEYVERSION,ANTENNAID,\n" +
                    "       OBUELECTRICAL,CONSUMETIME,PASSSTATE,INTERRUPTSIGNAL,SPECIALTYPE,SIGNSTATUS,VEHICLESIGNID,ENTOLLLANEID,\n" +
                    "       ENTOLLSTATIONHEX,ENTIME,ENLANETYPE,PASSID,LASTGANTRYHEX,LASTGANTRYTIME,OBUMAC,OBUISSUEID,OBUSN,OBUVERSION,\n" +
                    "       OBUSTARTDATE,OBUENDDATE,OBUSTATE,OBUVEHICLEPLATE,OBUVEHICLETYPE,VEHICLEUSERTYPE,VEHICLESEAT,AXLECOUNT,\n" +
                    "       TOTALWEIGHT,VEHICLELENGTH,VEHICLEWIDTH,VEHICLEHIGHT,CPUNETID,CPUISSUEID,CPUVEHICLEPLATE,CPUVEHICLETYPE,\n" +
                    "       CPUSTARTDATE,CPUENDDATE,CPUVERSION,CPUCARDTYPE,CPUCARDID,BALANCEBEFORE,BALANCEAFTER,GANTRYPASSCOUNT,\n" +
                    "       GANTRYPASSINFO,FEEPROVINFO,FEESUMLOCALBEFORE,FEESUMLOCALAFTER,FEECALCRESULT,FEEINFO1,FEEINFO2,\n" +
                    "       FEEINFO3,HOLIDAYSTATE,TRADERESULT,VERIFYCODE,VEHICLEPICID,VEHICLETAILPICID,MATCHSTATUS,VALIDSTATUS,\n" +
                    "       DEALSTATUS,RELATEDTRADEID,ALLRELATEDTRADEID,STATIONDBTIME,STATIONDEALTIME,STATIONVALIDTIME,\n" +
                    "       STATIONMATCHTIME,ENTOLLSTATIONNAME,DESCRIPTION,RECEIVETIME,PACKNAME,TRANSNO,RATECOMPUTE,updateuser,dealdesc,sysdate\n" +
                    "FROM grantry_trans_main_err WHERE haddeal in(1,3,4) and tradeId = ?");
        }else if(Constants.SQL_INST_CHARGE_BOUTLIST_DETAIL.equals(sqlId)){//MTC出口流水明细表
            sql.append("insert into tb_charge_boutlist_detail(SPARE5,ENPOINTID,EXPOINTID,ENTOLLSTATIONNAME,EXTOLLSTATIONNAME,PROVINCEINDEX,\n" +
                    "       ENTIME,EXTIME,INSERTTABLETIME,SEQ,ERRORTYPE,ERRORDETAIL,HADDEAL,DEALTIME,ERRORTOTAL,JSONNAME,LISTNO,\n" +
                    "       STATIONNO,SQUADDATE,PROVINCEID,TOLLSUPPORT,TOLLFEE,TOLLINTERVALSCOUNT,TOLLINTERVALSGROUP,TRANSTIMEGROUP,\n" +
                    "       CHARGEFEEGROUP,CHARGEDISCOUNTGROUP,RATEVERSIONGROUP,RECEIVETIME,PACKNAME,SECTIONGROUP,SECTIONNAME,\n" +
                    "       CHARGEPAYGROUP,TRANSFERTAG,SPARE1,SPARE2,SPARE3,SPARE4)\n" +
                    "SELECT SPARE5,ENPOINTID,EXPOINTID,ENTOLLSTATIONNAME,EXTOLLSTATIONNAME,PROVINCEINDEX,\n" +
                    "       ENTIME,EXTIME,INSERTTABLETIME,SEQ,ERRORTYPE,ERRORDETAIL,haddeal,DEALTIME,ERRORTOTAL,JSONNAME,LISTNO,\n" +
                    "       STATIONNO,SQUADDATE,PROVINCEID,TOLLSUPPORT,TOLLFEE,TOLLINTERVALSCOUNT,TOLLINTERVALSGROUP,TRANSTIMEGROUP,\n" +
                    "       CHARGEFEEGROUP,CHARGEDISCOUNTGROUP,RATEVERSIONGROUP,RECEIVETIME,PACKNAME,SECTIONGROUP,SECTIONNAME,\n" +
                    "       CHARGEPAYGROUP,TRANSFERTAG,SPARE1,SPARE2,SPARE3,SPARE4\n" +
                    "FROM boutlist_detail_err WHERE haddeal in(1,3,4) and listno LIKE ?");
        }else if(Constants.SQL_INST_CHARGE_ETCBOUTLIST_DETAIL.equals(sqlId)){//ETC出口流水明细表
            sql.append("insert into tb_charge_etcboutlist_detail(LISTNO,STATIONNO,SQUADDATE,PROVINCEID,TOLLSUPPORT,TOLLFEE,TOLLINTERVALSCOUNT,\n" +
                    "       TOLLINTERVALSGROUP,TRANSTIMEGROUP,CHARGEFEEGROUP,CHARGEDISCOUNTGROUP,RATEVERSIONGROUP,RECEIVETIME,PACKNAME,\n" +
                    "       SECTIONGROUP,SECTIONNAME,CHARGEPAYGROUP,TRANSFERTAG,SPARE1,SPARE2,SPARE3,SPARE4,SPARE5,ENPOINTID,EXPOINTID,\n" +
                    "       ENTOLLSTATIONNAME,EXTOLLSTATIONNAME,PROVINCEINDEX,ENTIME,EXTIME,INSERTTABLETIME,SEQ,ERRORTYPE,ERRORDETAIL,\n" +
                    "       HADDEAL,DEALTIME,ERRORTOTAL,JSONNAME)\n" +
                    "SELECT LISTNO,STATIONNO,SQUADDATE,PROVINCEID,TOLLSUPPORT,TOLLFEE,TOLLINTERVALSCOUNT,\n" +
                    "       TOLLINTERVALSGROUP,TRANSTIMEGROUP,CHARGEFEEGROUP,CHARGEDISCOUNTGROUP,RATEVERSIONGROUP,RECEIVETIME,PACKNAME,\n" +
                    "       SECTIONGROUP,SECTIONNAME,CHARGEPAYGROUP,TRANSFERTAG,SPARE1,SPARE2,SPARE3,SPARE4,SPARE5,ENPOINTID,EXPOINTID,\n" +
                    "       ENTOLLSTATIONNAME,EXTOLLSTATIONNAME,PROVINCEINDEX,ENTIME,EXTIME,INSERTTABLETIME,SEQ,ERRORTYPE,ERRORDETAIL,\n" +
                    "       haddeal,DEALTIME,ERRORTOTAL,JSONNAME\n" +
                    "FROM etcboutlist_detail_err WHERE haddeal in(1,3,4) and listno LIKE ?");
        }else if(Constants.SQL_DEL_CHARGE_BINLIST.equals(sqlId)){
            sql.append("delete from tb_charge_binlist where listNo = ? ");

        }else if(Constants.SQL_DEL_CHARGE_ETCBINLIST.equals(sqlId)){
            sql.append("delete from tb_charge_etcbinlist where listNo = ? ");

        }else if(Constants.SQL_DEL_CHARGE_BOUTLIST.equals(sqlId)){
            sql.append("delete from tb_charge_boutlist where listNo = ? ");

        }else if(Constants.SQL_DEL_CHARGE_ETCBOUTLIST.equals(sqlId)){
            sql.append("delete from tb_charge_etcboutlist where listNo = ? ");

        }else if(Constants.SQL_DEL_CHARGE_GRANTRY_MAIN.equals(sqlId)){
            sql.append("delete from tb_charge_grantry_trans_main where tradeId = ? ");

        }else if(Constants.SQL_DEL_CHARGE_BOUTLIST_DETAIL.equals(sqlId)){
            sql.append("delete from tb_charge_boutlist_detail where listNo like ? ");

        }else if(Constants.SQL_DEL_CHARGE_ETCBOUTLIST_DETAIL.equals(sqlId)){
            sql.append("delete from tb_charge_etcboutlist_detail where listNo like ? ");

        }else if(Constants.SQL_UPD_CHARGE_BINLIST.equals(sqlId)){
            sql.append("update tb_charge_binlist set haddeal = 1 where listNo = ?");

        }else if(Constants.SQL_UPD_CHARGE_ETCBINLIST.equals(sqlId)){
            sql.append("update tb_charge_etcbinlist set haddeal = 1 where listNo = ? ");

        }else if(Constants.SQL_UPD_CHARGE_BOUTLIST.equals(sqlId)){
            sql.append("update tb_charge_boutlist set haddeal = 1 where listNo = ? ");

        }else if(Constants.SQL_UPD_CHARGE_ETCBOUTLIST.equals(sqlId)){
            sql.append("update tb_charge_etcboutlist set haddeal = 1 where listNo = ? ");

        }else if(Constants.SQL_UPD_CHARGE_GRANTRY_MAIN.equals(sqlId)) {
            sql.append("update tb_charge_grantry_trans_main set haddeal = 1 where tradeId = ? ");
        }
        return sql.toString();
    }

}

/**
 * 多线程实现类
 */
class Task implements Callable<String> {
    private String listNo;
    private JSONArray jsonArray;
    private String filename;
    private KafkaProducer producer;


    public Task(KafkaProducer producer,String listNo,JSONArray jsonArray,String filename) {
        this.producer = producer;
        this.listNo = listNo;
        this.jsonArray = jsonArray;
        this.filename = filename;
    }

    @Override
    public String call() throws Exception {
        try {
            long start = System.currentTimeMillis();
            ApplicationContext ac = SpringContextHolder.getApplicationContext();
            ExceptionDistributeService service = (ExceptionDistributeService)ac.getBean("exceptionDistributeService");
            service.sendToKafka(producer,jsonArray,filename);
            XxlJobLogger.log("发送成功包名："+filename);
            return "success";

            //int reponseStatus = service.sendPlatform(jsonArray,filename);//调用中台http接口
            //XxlJobLogger.log("接口响应状态:" + reponseStatus+",共耗时："+(System.currentTimeMillis()-start)/1000+"秒");
/*            if(200 == reponseStatus){
                return "success";
            }else{
                return listNo;
            }*/
            //XxlJobLogger.log("共耗时："+(System.currentTimeMillis()-start)/1000+"秒");
        } catch (Exception e) {
            //当发送失败时，返回主键号用于将该流水记录的状态重新改为“1-已治理”等待重新发送
            XxlJobLogger.log("发送失败流水号："+listNo);
            e.printStackTrace();
            return listNo;
        }
    }
}


